package com.hjl.demo.kafka;

import com.alibaba.ververica.cdc.debezium.DebeziumDeserializationSchema;
import io.debezium.data.Envelope;
import net.minidev.json.JSONObject;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.util.Collector;
import org.apache.kafka.connect.data.Field;
import org.apache.kafka.connect.data.Schema;
import org.apache.kafka.connect.data.Struct;
import org.apache.kafka.connect.source.SourceRecord;

import java.util.List;

public class CustomerDeserialization implements DebeziumDeserializationSchema<String> {
   /*封装数据的格式
   {
        "database":"",
        "tableName":"",
        "type":"c u d",
        "before":{"id":"","tm_name":""....},
        "after":{"id":"","tm_name":""....}
        //"ts":  befor，after 就是说 数据里面有 Eventtime
   }
    */
    public void deserialize(SourceRecord sourceRecord, Collector<String> collector) throws Exception {
        //1.创建Json对象用于存储最终数据
        JSONObject result = new JSONObject();

        //2.获取库名和表名
        String topic = sourceRecord.topic();
        String[] fields = topic.split("\\.");//转义
        String database = fields[1];
        String tableName = fields[2];

        Struct value = (Struct) sourceRecord.value();
        //3.获取“befor”数据
        Struct before = value.getStruct("before");//
        JSONObject beforeJson = new JSONObject();
        if(before !=null) {
            Schema beforeSchema = before.schema();//字段
            List<Field> beforeFields = beforeSchema.fields();//存储在列表
            for (Field field : beforeFields) {  //field 就是 id  name    beforeValue就是id的值，name的值
                Object beforeValue = before.get(field);
                beforeJson.put(field.name(), beforeValue);
            }
        }
        //4.获取“after”数据
        Struct after = value.getStruct("after");//
        JSONObject afterJson = new JSONObject();
        if(after !=null) {
            Schema afterSchema = after.schema();//字段
            List<Field> afterFields = afterSchema.fields();//存储在列表
            for (Field field : afterFields) {  //field 就是 id  name    afterValue就是id的值，name的值
                Object afterValue = after.get(field);
                afterJson.put(field.name(), afterValue);
            }
        }
        //5.获取操作类型 CREATE UPDATE DELETE
        Envelope.Operation operation = Envelope.operationFor(sourceRecord);
        String type = operation.toString().toLowerCase();
        if("create".equals(type)){
            type = "insert";
        }

        //6.将字段写入Json对象
        result.put("database", database);
        result.put("tableName", tableName);
        result.put("before", beforeJson);
        result.put("after", afterJson);
        result.put("type",type);

        //7.输出数据
        collector.collect(result.toString());
    }

    public TypeInformation<String> getProducedType() {
        return BasicTypeInfo.STRING_TYPE_INFO;
    }
}
